package plgmqtt

import (
	"dgo/MyProj/PrjHttpUtils"
	"dgo/goutils/utils"
	"dgo/plg/plgbase"
	"fmt"
	"gitee.com/ymofen/gobase/golog"
	"gitee.com/ymofen/supervalue"
	mqtt "github.com/eclipse/paho.mqtt.golang"
	"strings"
	"sync/atomic"
	"time"
)

type PlgExMqtt struct {
	send_err_n  uint32
	send_n      uint32
	push_fail_n uint32

	PlgBase
	host     string
	user     string
	pass     string
	clientid string

	clinetOptions *mqtt.ClientOptions
	client        mqtt.Client
	timer         *utils.Timer

	data_ex *utils.SubscribeCenter // 同步播发数据

	/* 订阅列表 */
	sublst utils.GSyncMapGroup

	monitor_send *utils.MonitorSpeedSizeRec
	monitor_recv *utils.MonitorSpeedSizeRec
	monitor_push *utils.MonitorSpeedSizeRec

	last_warning string
	UseCache     bool
	cachelst     *utils.QueueTask

	onRecvEvent *utils.Dispatch
}

func (this *PlgExMqtt) innerConfig() {
	cltOptions := mqtt.NewClientOptions()
	cltOptions.AddBroker(this.host).SetAutoReconnect(true)
	if len(this.user) > 0 {
		cltOptions.SetUsername(this.user).SetPassword(this.pass)
	}
	if len(this.clientid) > 0 {
		cltOptions.SetClientID(this.clientid)
	}

	cltOptions.SetOnConnectHandler(func(client mqtt.Client) {
		this.sublst.Range(func(key, value interface{}) bool {
			channelid := key.(string)
			this.innerSubscribe(channelid)
			return true
		})

		golog.Infof("建立(%s)连接成功", this.host)
	})
	cltOptions.SetConnectionLostHandler(func(client mqtt.Client, err error) {
		golog.Infof("连接丢失:%s", err.Error())
	})
	cltOptions.SetConnectTimeout(time.Duration(10) * time.Second)
	this.clinetOptions = cltOptions
}

func (this *PlgExMqtt) AddListener(groupid, id interface{}, ls utils.Listener_CallBack) {
	if groupid == nil {
		this.onRecvEvent.AddListener(id, ls)
	} else if groupstr, ok := groupid.(string); ok {
		if groupstr == "onrecv" {
			this.onRecvEvent.AddListener(id, ls)
		}
	}
}

func (this *PlgExMqtt) RemoveListener(groupid, id interface{}) {
	if groupid == nil {
		this.onRecvEvent.RemoveListener(id)
	} else if groupstr, ok := groupid.(string); ok {
		if groupstr == "onrecv" {
			this.onRecvEvent.RemoveListener(id)
		}
	}
}

func (this *PlgExMqtt) innerCheckConnect() {
	if this.ReleaseFlag == 1 {
		if this.client != nil {
			this.client.Disconnect(1)
			this.client = nil
		}
		return
	}

	if len(this.host) == 0 {
		return
	}
	if this.client != nil {
		return
	}

	clt := mqtt.NewClient(this.clinetOptions)
	token := clt.Connect()

	if token.WaitTimeout(time.Second * 5); token.Error() != nil {
		golog.Warnf("连接(%s)异常:%s", this.host, token.Error().Error())
		return
	}
	if clt.IsConnected() {
		this.client = clt
		return
	} else {
		golog.Warnf("连接(%s)超时", this.host)
		return
	}
}

/*
**

	{
	  "id":"mqtt-ex",
	  "type":"plg.mqtt.ex",
	  "conf":{
		"host":"tcp://139.199.7.76:52883",
	    "user":"",
	    "pass":"",
	    "usecache":false
	   }
	}
*/
func (this *PlgExMqtt) ReloadConf(conf *wrapper.SuperValue) error {

	if this.onRecvEvent == nil {
		this.onRecvEvent = utils.NewDispatch(this.PlugId)
	}

	if this.monitor_send == nil {
		this.monitor_send = utils.NewMonitorSpeedSizeRec()
		this.monitor_recv = utils.NewMonitorSpeedSizeRec()
		this.monitor_push = utils.NewMonitorSpeedSizeRec()
		utils.Monitor.AddMonitorRec(this.monitor_send, this.monitor_send)
		utils.Monitor.AddMonitorRec(this.monitor_recv, this.monitor_recv)
		utils.Monitor.AddMonitor(this.monitor_push)
	}

	isChanged := false

	conf.CheckExistsStrFunc("user", func(strv string) {
		strv = utils.ReplacePlaceholder(strv, "$$", "$$")
		if strv != this.user {
			this.user = strv
			isChanged = true
		}
	})

	conf.CheckExistsStrFunc("pass", func(strv string) {
		strv = utils.ReplacePlaceholder(strv, "$$", "$$")
		if strv != this.pass {
			this.pass = strv
			isChanged = true
		}
	})

	conf.CheckExistsStrFunc("host", func(strv string) {
		strv = utils.ReplacePlaceholder(strv, "$$", "$$")
		strv = plgbase.ReplacePlaceholderForFileName(strv)
		if this.host != strv {
			this.host = strv
			isChanged = true
		}
	})

	conf.CheckExistsStrFunc("usecache", func(strv string) {
		v := utils.TryStrToBool(strv, false)
		if this.UseCache != v {
			if v {
				if this.cachelst == nil {
					this.cachelst = utils.NewQueueTaskFunc()
					this.cachelst.ConfigMaxQueueSize(20480)
					this.cachelst.ConfigWorkNum(3, 5)
					this.cachelst.ConfigTaskCallBackFunc(func(args ...interface{}) {
						if len(args) == 2 {
							topic := args[0].(string)
							err := this.innerSend(topic, args[1])
							if err != nil {
								this.last_warning = fmt.Sprintf("[%s][%s]推送数据异常:%s", utils.NowString(), topic, err.Error())
							}
						}
					})
					this.cachelst.Start()

					golog.Infof("开启缓存推送模式!")
				}
			}
			this.UseCache = v
			if !v {
				if this.cachelst != nil {
					this.cachelst.Stop()
				}
			}
		}
	})

	if this.timer == nil {
		this.timer = utils.NewTimer()
		this.timer.OnTimer = func(tag interface{}) {
			this.innerCheckConnect()
		}
		this.timer.TrigID = this
		this.timer.ConfigInterval(time.Second*30, 0)
	}

	if isChanged {
		utils.DefaultWorkers().PostTaskFunc(this, func() {
			if this.client != nil {
				this.client.Disconnect(1)
			}

			if len(this.host) > 0 {
				this.innerConfig()
				this.innerCheckConnect()
			} else {
				this.client.Disconnect(1)
				this.client = nil
			}
		})
	}

	if this.data_ex == nil {
		this.data_ex = utils.NewSubscribeCenter()
		this.data_ex.OnUnSubChannel = func(channelid string, n int) {
			if n == 0 {
				if channelid == "*" {
					return
				}
				utils.DefaultWorkers().PostTaskFunc(this, func() {
					this.innerUnsubscribe(channelid)
				})
			}
		}
	}
	return nil
}

func (this *PlgExMqtt) OnObjectNotify(sender interface{}, notifytype int, data interface{}) (rval interface{}, err error) {
	if notifytype == utils.NOTIFY_TYPE_SETCONF {
		if conf, ok := data.(*wrapper.SuperValue); ok {
			err = this.ReloadConf(conf)
		} else {
			err = fmt.Errorf("非法的JSON配置")
		}
	} else if notifytype == utils.NOTIFY_TYPE_RELEASE {
		this.ReleaseFlag = 1

		if this.monitor_send != nil {
			utils.Monitor.RemoveMonitorRec(this.monitor_send)
			utils.Monitor.RemoveMonitorRec(this.monitor_recv)
			utils.Monitor.RemoveMonitorRec(this.monitor_push)
		}

		if this.timer != nil {
			this.timer.Close()
		}

		clt := this.client
		if clt != nil {
			clt.Disconnect(1)
			this.client = nil
		}
	}
	return
}

func (this *PlgExMqtt) onMqttRecv(client mqtt.Client, msg mqtt.Message) {
	if utils.GoFunCatchException {
		defer utils.PanicHandler()
	}
	buf := msg.Payload()
	this.monitor_recv.DeltaBuf(buf)
	this.data_ex.DispatchData(msg.Topic(), this, buf, nil)
	this.onRecvEvent.Dispatch(this, msg.Topic(), buf)
}

/*
*
通道第一次创建的时候订阅
*/
func (this *PlgExMqtt) innerSubscribe(channelid string) bool {
	wlt := this.client
	if wlt == nil {
		return false
	}
	recvToken := wlt.Subscribe(channelid, 0, this.onMqttRecv)
	return recvToken.Wait()
}

func (this *PlgExMqtt) innerUnsubscribe(channelid string) bool {
	wlt := this.client
	if wlt == nil {
		return false
	}
	wlt.Unsubscribe(channelid)
	this.sublst.RemoveEx(channelid)
	return true
}

func (this *PlgExMqtt) SubscribeChannel(channelid string, id interface{}, cb utils.SubscribeCallback) {
	this.data_ex.SubscribeChannel(channelid, id, cb)

	e1 := utils.DefaultWorkers().PostTaskFunc(this, func() {
		if this.sublst.Get(channelid, utils.ZeroTime).(time.Time).IsZero() {
			if this.innerSubscribe(channelid) {
				this.sublst.Set(channelid, time.Now())
				this.data_ex.DispatchData(channelid, id, nil, 1) // 订阅成功
			} else {
				this.data_ex.DispatchData(channelid, id, nil, 0) // 订阅失败
				golog.Warnf(fmt.Sprintf("[%s]订阅失败", channelid))
			}
		} else {
			this.data_ex.DispatchData(channelid, id, nil, 1) // 订阅成功
		}
	})
	if e1 != nil {
		this.data_ex.DispatchData(channelid, id, nil, 0) // 订阅失败
		golog.Warnf(fmt.Sprintf("[%s]投递订阅信息失败:%s", channelid, e1.Error()))
	}
}

func (this *PlgExMqtt) UnsubscribeChannel(channelid string, id interface{}) bool {
	return this.data_ex.UnsubscribeChannel(channelid, id)
}

func (this *PlgExMqtt) SendKVs(topic string, kv ...interface{}) error {
	var data interface{}
	l := len(kv)
	if l == 1 {
		data = kv[0]
		if buf, ok := data.([]byte); ok {
			this.monitor_push.DeltaBuf(buf)
		}
	} else {
		json := wrapper.NewSVObject()
		json.SetKeyValues(kv...)
		data = json.String()
	}

	if this.UseCache {
		err := this.cachelst.Push(topic, data)
		if err != nil {
			atomic.AddUint32(&this.push_fail_n, 1)
			this.last_warning = fmt.Sprintf("[%s][%s]压入队列异常:%s", utils.NowString(), topic, err.Error())
		}
		return err
	} else {
		err := this.innerSend(topic, data)
		if err != nil {
			this.last_warning = fmt.Sprintf("[%s][%s]推送数据异常:%s", utils.NowString(), topic, err.Error())
		}
		return err
	}
}

func (this *PlgExMqtt) innerSend(topic string, data interface{}) error {
	clt := this.client
	if clt == nil {
		return fmt.Errorf("Redis连接尚未进行创建")
	}

	atomic.AddUint32(&this.send_n, 1)

	if buf, ok := data.([]byte); ok {
		token := clt.Publish(topic, 0, false, buf)
		if !token.Wait() {
			atomic.AddUint32(&this.send_err_n, 1)
			return fmt.Errorf("发送失败")
		}
		this.monitor_send.DeltaBuf(buf)
	} else if str, ok := data.(string); ok {
		buf := []byte(str)
		token := clt.Publish(topic, 0, false, buf)
		if !token.Wait() {
			atomic.AddUint32(&this.send_err_n, 1)
			return fmt.Errorf("发送失败")
		}
		this.monitor_send.DeltaBuf(buf)
	} else if json, ok := data.(*wrapper.SuperValue); ok {
		str := json.String()
		token := clt.Publish(topic, 0, false, str)
		if !token.Wait() {
			atomic.AddUint32(&this.send_err_n, 1)
			return fmt.Errorf("发送失败")
		}
		this.monitor_send.Delta(int64(len(str)))
	} else {
		atomic.AddUint32(&this.send_err_n, 1)
		return fmt.Errorf("未知的数据类型:%v", data)
	}
	return nil
}

func (this *PlgExMqtt) DispatchData(channelid string, sender interface{}, data interface{}, tag interface{}) (err error) {
	return this.SendKVs(channelid, data)
}

func (this *PlgExMqtt) OnHttpStatus(rec *PlgHttpRec) (rval interface{}, errcode int, err error) {
	itype := rec.QryAsInt("type", 0)
	if itype == 99 {
		json := wrapper.NewSVObject()
		json.SetKeyValues("send-info", this.monitor_send.Info(),
			"recv-info", this.monitor_recv.Info(),
			"push-info", this.monitor_push.Info(),
			"send-err-n", this.send_err_n,
			"send-n", this.send_n,
			"last-warning", this.last_warning,
			"sub-n", this.sublst.Count(),
			"use-cache", this.UseCache,
		)
		if this.cachelst != nil {
			json.SetKeyValue("cache", fmt.Sprintf("n:%d, pushfail:%d", this.cachelst.Count(), this.push_fail_n))
		}

		PrjHttpUtils.JSON2TableRec(rec.RespW, "", json.AsJSONString(false))
	} else if itype == 1 {
		var sb utils.BytesBuilder
		searchval := rec.QryAsString("searchval", "")
		n := 0
		this.sublst.Range(func(key, value interface{}) bool {
			n++
			channelid := key.(string)
			if len(searchval) == 0 || strings.Contains(channelid, searchval) {
				sb.Appendf("%d: [%s][start:%s]\r\n", n, channelid, utils.DateTimeString3(value.(time.Time)))
			}
			return true
		})
		rec.RespW.Write(sb.Bytes())
	} else if itype == 2 {
		rec.ResponseText(this.data_ex.StatusString())
	} else if itype == 3 {
		//rec.ResponseSyncListN(this.warnlst, rec.QryAsString("searchval", ""), rec.QryAsInt("n", 0))
	} else {
		var sb utils.BytesBuilder
		PrjHttpUtils.AppendIndexBody(&sb, "status", "status?type=99")
		PrjHttpUtils.AppendIndexBody(&sb, "subscribe-list", "status?type=1&searchval=")
		PrjHttpUtils.AppendIndexBody(&sb, "data_ex-status", "status?type=2")
		//PrjHttpUtils.AppendIndexBody(&sb, "warn-list", "status?type=3&searchval=")
		PrjHttpUtils.ResponseBodyHtml(rec.RespW, sb.String())
	}

	return
}

func NewPlgMqttEx() *PlgExMqtt {
	obj := &PlgExMqtt{}
	return obj
}

func CreatePlgExMqttFunc(args ...interface{}) (rval interface{}, err error) {
	obj := &PlgExMqtt{}
	if len(args) > 0 {
		if str, ok := args[0].(string); ok {
			obj.PlugId = str
		}
	}
	rval = obj
	return
}
